/*
 *   Copyright 2021 zzh
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.aduib.boot.common.buffer;

import com.aduib.boot.common.concurrent.JctoolsHelper;
import com.aduib.boot.common.log.LOG;

import java.util.Objects;
import java.util.Set;

/**
 * ChunkList维护着一个指向一串Chunk的头结点，访问策略由minUsage，maxUsage决定
 *
 * @description: ByteBufferChunkList
 * @author: zzh
 * @date: 2021/12/1 15:05
 */
public final class BufferChunkList {

  private final int minUsage;

  private final int maxUsage;

  private final int maxCapacity;
  private final int freeMinThreshold;
  private final int freeMaxThreshold;
  private final BufferPool bufferPool;
  Set<BufferChunk> byteBufferChunks;
  BufferChunkList prevList;
  BufferChunkList nextList;
  BufferChunk head;

  public BufferChunkList(int minUsage, int maxUsage, int chunkSize, BufferPool bufferPool) {
    this.minUsage = minUsage;
    this.maxUsage = maxUsage;
    this.bufferPool = bufferPool;
    byteBufferChunks = JctoolsHelper.nonBlockingHashSet();
    maxCapacity = calculateMaxCapacity(minUsage, chunkSize);
    //    createChunkList(chunkSize, pageSize, numOfChunks, bufferPool);

    freeMinThreshold =
        (maxUsage == 100) ? 0 : (int) (chunkSize * (100.0 - maxUsage + 0.99999999) / 100L);
    freeMaxThreshold =
        (minUsage == 100) ? 0 : (int) (chunkSize * (100.0 - minUsage + 0.99999999) / 100L);
  }

  private static int calculateMaxCapacity(int minUsage, int chunkSize) {
    minUsage = Math.max(1, minUsage);

    if (minUsage == 100) {
      // If the minUsage is 100 we can not allocate anything out of this list.
      return 0;
    }

    // Calculate the maximum amount of bytes that can be allocated from a PoolChunk in this
    // PoolChunkList.
    //
    // As an example:
    // - If a PoolChunkList has minUsage == 25 we are allowed to allocate at most 75% of the
    // chunkSize because
    //   this is the maximum amount available in any PoolChunk in this PoolChunkList.
    return (int) (chunkSize * (100L - minUsage) / 100L);
  }

  private void createChunkList(
      int chunkSize, int pageSize, int numOfChunks, BufferPool bufferPool) {
    for (int i = 0; i < numOfChunks; i++) {
      BufferChunk chunk =
          new BufferChunk(pageSize, chunkSize, bufferPool, BufferUtils.allocateBuffer(chunkSize));
      byteBufferChunks.add(chunk);
    }
  }

  public BufferChunk getIndex(Buffer buffer) {
    for (BufferChunk byteBufferChunk : byteBufferChunks) {
      if (byteBufferChunk.isInThisChunk(buffer)) {
        return byteBufferChunk;
      }
    }
    return null;
  }

  Buffer allocate(int reqCapacity) {
    for (BufferChunk cur : byteBufferChunks) {
      Buffer buf = cur.allocateRun(reqCapacity);
      if (Objects.nonNull(buf)) {
        final int usage = cur.usage();
        if (usage >= maxUsage) {
          BufferChunkList next = nextList;
          BufferChunkList current = this;
          while (next != null) {
            current.byteBufferChunks.remove(cur);
            next.byteBufferChunks.add(cur);
            if (next.maxUsage > usage) {
              break;
            }
            current = next;
            next = next.nextList;
          }
        }
        return buf;
      }
    }
    return null;
  }

  /**
   * 申请buffer
   *
   * @param buf
   * @param reqCapacity
   * @param threadCache
   * @return
   */
  boolean allocate(PooledBuffer buf, int reqCapacity, int sizeIdx, PoolThreadCache threadCache) {
    int normCapacity = ((BufferPoolArena) bufferPool).sizeIdx2size(sizeIdx);
    if (normCapacity > maxCapacity) {
      // Either this PoolChunkList is empty or the requested capacity is larger then the capacity
      // which can
      // be handled by the PoolChunks that are contained in this PoolChunkList.
      return false;
    }
    for (BufferChunk cur = head; cur != null; cur = cur.next) {
      if (cur.allocate(buf, reqCapacity, sizeIdx, threadCache)) {
        if (cur.freeBytes <= freeMinThreshold) {
          remove(cur);
          nextList.add(cur);
          //          byteBufferChunks.add(cur);
        }
        return true;
      }
    }
    return false;
  }

  boolean free(Buffer buffer) {
    BufferChunk cur = getIndex(buffer);
    if (cur == null) {
      LOG.i("not in this list!");
      return false;
    }
    cur.freeByteBuffer(buffer);
    final int usage = cur.usage();
    if (usage < minUsage) {
      BufferChunkList prev = prevList;
      BufferChunkList current = this;
      while (prev != null) {
        current.byteBufferChunks.remove(cur);
        prev.byteBufferChunks.add(cur);
        if (prev.minUsage < usage) {
          break;
        }
        current = prev;
        prev = prev.prevList;
      }
    }
    return true;
  }

  boolean free(long handle, int maxLength, BufferChunk chunk, Buffer nioBuffer) {
    chunk.free(handle, maxLength, nioBuffer);
    if (chunk.freeBytes > freeMaxThreshold) {
      remove(chunk);
      // Move the PoolChunk down the PoolChunkList linked-list.
      return move0(chunk);
    }
    return true;
  }

  private boolean move(BufferChunk chunk) {
    assert chunk.usage() < maxUsage;

    if (chunk.freeBytes > freeMaxThreshold) {
      // Move the BufferChunk down the BufferChunkList linked-list.
      return move0(chunk);
    }

    // BufferChunk fits into this BufferChunkList, adding it here.
    add0(chunk);
    return true;
  }

  /**
   * Moves the {@link BufferChunk} down the {@link BufferChunkList} linked-list so it will end up in
   * the right {@link BufferChunkList} that has the correct minUsage / maxUsage in respect to {@link
   * BufferChunk#usage()}.
   */
  private boolean move0(BufferChunk chunk) {
    if (prevList == null) {
      // There is no previous BufferChunkList so return false which result in having the BufferChunk
      // destroyed and
      // all memory associated with the BufferChunk will be released.
      assert chunk.usage() == 0;
      return false;
    }
    return prevList.move(chunk);
  }

  void add(BufferChunk chunk) {
    if (chunk.freeBytes <= freeMinThreshold) {
      nextList.add(chunk);
      return;
    }
    add0(chunk);
  }

  /** Adds the {@link BufferChunk} to this {@link BufferChunkList}. */
  void add0(BufferChunk chunk) {
    chunk.parent = this;
    if (head == null) {
      head = chunk;
      chunk.prev = null;
      chunk.next = null;
    } else {
      chunk.prev = null;
      chunk.next = head;
      head.prev = chunk;
      head = chunk;
    }
  }

  private void remove(BufferChunk cur) {
    if (cur == head) {
      head = cur.next;
      if (head != null) {
        head.prev = null;
      }
    } else {
      BufferChunk next = cur.next;
      cur.prev.next = next;
      if (next != null) {
        next.prev = cur.prev;
      }
    }
  }
}
